package com.junmo.www.RxJava;


import io.reactivex.*;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import org.testng.annotations.Test;

public class BackPressureTest {
    @Test
    public void  test(){

    }
    @Test
    public void  test01() throws InterruptedException {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                System.out.println("-----数据发送------"+Thread.currentThread().getName());
                for(int i=0;i<130;i++){
                    e.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.computation())
                .observeOn(Schedulers.single())

                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("-----数据接收------"+Thread.currentThread().getName());
                        System.out.println (integer);
                    }
        });
        Thread.sleep(20000L);
    }


}
